#include "thread_pool.h"

using namespace std;

namespace cppdt {

    ThreadPool::ThreadPool()
        : mMinNum(0)
        , mMaxNum(0)
        , mTasks() {
    }

    ThreadPool::~ThreadPool() {
    }

    void ThreadPool::Init(size_t minThreadNum, size_t maxThreadNum) {
        assert(0 == mMinNum && 0 == mMaxNum);
        assert(minThreadNum > 0);
        assert(maxThreadNum >= minThreadNum);

        mMinNum = minThreadNum;
        mMaxNum = maxThreadNum;

        std::unique_lock<std::mutex> ul(mMutex);
        for (size_t i = 0; i < minThreadNum; ++i) {
            mThreads.push_back(new thread(&ThreadPool::OnThread, this));
        }
    }

    void ThreadPool::Join() {
        vector<ThreadInfo> tmp;
        {
            std::unique_lock<std::mutex> ul(mMutex);
            for (vector<ThreadInfo>::iterator it = mThreads.begin(); it != mThreads.end(); ++it) {
                mTasks.Enqueue(ThreadData());
                tmp.push_back(*it);
            }
            mThreads.clear();
        }
        for (vector<ThreadInfo>::iterator it = tmp.begin(); it != tmp.end(); ++it) {
            it->mThread->join();
            delete it->mThread;
        }
    }

    void ThreadPool::Execute(IThreadTask* task, void* param) {
        assert(task);

        vector<ThreadInfo> tmp;
        {
            std::unique_lock<std::mutex> ul(mMutex);
            if (mTasks.Enqueue(ThreadData(task, param)) > mMinNum) {
                if (mThreads.size() >= mMinNum && mThreads.size() < mMaxNum) {
                    mThreads.push_back(new thread(&ThreadPool::OnThread, this));
                }
            }

            vector<ThreadInfo>::iterator it = mThreads.begin();
            while (it != mThreads.end()) {
                if (it->mEnable) {
                    ++it;
                } else {
                    tmp.push_back(*it);
                    it = mThreads.erase(it);
                }
            }
        }
        for (vector<ThreadInfo>::iterator it = tmp.begin(); it != tmp.end(); ++it) {
            it->mThread->join();
            delete it->mThread;
        }
    }

    void ThreadPool::OnThread() {
        ThreadData data;
        while (1) {
            if (mTasks.Dequeue(data, 300000)) {
                if (data.mTask) {
                    data.mTask->OnThread(*this, data.mParam);
                } else {
                    break;
                }
            } else {
                std::unique_lock<std::mutex> ul(mMutex);
                size_t count = 0;
                vector<ThreadInfo>::iterator curr = mThreads.end();
                for (vector<ThreadInfo>::iterator it = mThreads.begin(); it != mThreads.end(); ++it) {
                    if (it->mEnable)
                        ++count;

                    if (this_thread::get_id() == it->mThread->get_id())
                        curr = it;
                }
                if (count > mMinNum && curr != mThreads.end()) {
                    curr->mEnable = false;
                    break;
                }
            }
        }
    }

}
